package me.shiqiang.im.listeners;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.fasterxml.jackson.core.JsonProcessingException;
import me.shiqiang.im.enums.MessageCType;
import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.msg.DistributeMsgBody;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.NewMsg;
import me.shiqiang.im.msg.SndMsg;
import me.shiqiang.im.redis.RedisKey;
import me.shiqiang.im.service.GroupClient;
import me.shiqiang.im.service.MsgService;
import me.shiqiang.im.statistics.MsgCount;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.util.RedisPipelineProcessor;
import me.shiqiang.im.util.SnowFlake;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author cuishiqiang
 * @date 2018/1/12 下午1:43
 */
@Component("SndMsgListener")
public class SndMsgListener extends ConnectServerListener {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public String eventName() {
        return MessageCmd.SND_MSG;
    }

    @Autowired
    StringRedisTemplate redisTemplate;

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Autowired
    GroupClient groupClient;

    @Autowired
    SnowFlake snowFlake;

    @Value("${send.topic}")
    private String topic;

    @Autowired
    Producer producer;

    @Autowired
    MsgService msgService;

    @Value("${server.id}")
    String serverId;

    @Override
    public void call(Object... objects) {
        JSONObject obj = (JSONObject) objects[0];
//        boolean remote = (boolean) objects[1];
        try {
            SndMsg sndMsg = JsonUtils.getInstance().fromJson(obj.toString(), SndMsg.class);
            logger.info("SndMsgListener收到消息：{}",sndMsg.toString());
            queue.add(connection -> {
                long syncKey = snowFlake.nextId();
                MsgBody msgBody = sndMsg.newMsgBody(syncKey, 0);
                try {
                    String json = JsonUtils.getInstance().toJson(msgBody);
                    String cType = String.valueOf(sndMsg.getcType());
                    if (MessageCType.P2P.equals(cType)) {
                        connection.zAdd(RedisKey.clientMsgQueue(sndMsg.getFrom().toString()).getBytes(), syncKey, json.getBytes());
                        NewMsg fromNewMsg = new NewMsg(sndMsg.getFrom(), msgBody.getSyncKey(),msgBody);

                        socket.emit(MessageCmd.NEW_MSG, new JSONObject(JsonUtils.getInstance().toJson(fromNewMsg)));
                        String serverId = msgService.getServerId(sndMsg.getTo());
                        if (this.serverId.equals(serverId)) {
                            connection.zAdd(RedisKey.clientMsgQueue(sndMsg.getTo().toString()).getBytes(), syncKey, json.getBytes());
                            NewMsg toNewMsg = new NewMsg(sndMsg.getTo(), msgBody.getSyncKey(),msgBody);
                            socket.emit(MessageCmd.NEW_MSG, new JSONObject(JsonUtils.getInstance().toJson(toNewMsg)));
                        } else {
                            sendMqMsg(cType, sndMsg.getTo(), syncKey, json);
                        }
                        MsgCount.getInstance().p2pIncrement();
                    } else if (MessageCType.GROUP.equals(cType) || MessageCType.SERVICE_PLATFORM.equals(cType)) {
                        //此处将Long[] at 原封不动传给分发端
                        DistributeMsgBody distributeMsgBody = sndMsg.newMsgBody(syncKey);
                        json = JsonUtils.getInstance().toJson(distributeMsgBody);
                        logger.info("distributeMsgbody: " + json);
                        //添加对服务平台类型的支持,统一当群聊消息处理
                        boolean muted = groupClient.isMuted(sndMsg.getTo(), sndMsg.getFrom());
                        boolean allMutedAndIsNormal = groupClient.allMutedAndIsNormal(sndMsg.getTo(), sndMsg.getFrom());
                        if (allMutedAndIsNormal) {
                            return;
                        }
                        if (muted){
                            return;
                         }
                        sendMqMsg(MessageCType.GROUP, sndMsg.getTo(), syncKey, json);
                        MsgCount.getInstance().groupIncrement();
                    }
                } catch (JsonProcessingException e) {
                    logger.error(String.format("消息格式转换异常，消息: %s", msgBody.toString()), e);
                }
            });
        } catch (IOException e) {
            logger.error(String.format("消息格式转换异常，消息: %s", obj.toString()), e);
        }
    }

    private void sendMqMsg(String cType, Long to, Long syncKey, String body) {
        Message msg = new Message();
        msg.setTopic(topic);
        msg.setTag(cType); //cType  p2p 或 group
        msg.setBody(body.getBytes());
        Properties properties = new Properties();
        properties.setProperty("to", to.toString());
        properties.setProperty("syncKey", syncKey.toString());
        msg.setUserProperties(properties);
        producer.sendOneway(msg);
    }


}
